package coroutines

import kotlinx.coroutines.NonCancellable.cancel
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.suspendCancellableCoroutine

object FlowTest {

    fun test(){
        runBlocking {
           onCompletionTest()
        }
    }

    private suspend fun emitTest(){
        flow {
            emit(1)
            emit(2)
            emit(3)
            emit(4)
            emit(5)
        }.filter { it > 2 }
            .map { it * 2 }
            .take(2)
            .collect{
                println(it)
            }
    }

    private suspend fun onCompletionTest(){
        flowOf(1, 2, 3, 4, 5)
            .onCompletion { println("onCompletion:$it") } // 注意这里
            .filter {
                println("filter: $it")
                it > 2
            }
            .take(2)
            .collect {
                println("collect: $it")
            }

        flow {
            emit(1)
            emit(2)
            emit(3)
        }.onCompletion { println("onCompletion first: $it") }
            .collect {
                println("collect: $it")
                if (it == 2) {
                    cancel()           // 1
                    println("cancel")
                }
            }

        delay(100L)

        flowOf(4, 5, 6)
            .onCompletion { println("onCompletion second: $it") }
            .collect {
                println("collect: $it")
                // 仅用于测试，生产环境不应该这么创建异常
                throw IllegalStateException() // 2
            }
    }
}